package com.yand.redis;


import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.TypeReference;
import redis.clients.jedis.Jedis;

import java.lang.reflect.Type;
import java.util.Set;
import java.util.UUID;
import java.util.concurrent.TimeUnit;

/**
 * java 版本的 redis 的延迟队列
 */
public class RedisDelayingQueue<T> {

    static class TaskItem<T>{
        public String id;
        public T msg ;
    }


    private Type TaskType = new TypeReference(){}.getType();

    private Jedis jedis;
    private String  queueKey;

    public RedisDelayingQueue(Jedis jedis, String queueKey) {
        this.jedis = jedis;
        this.queueKey = queueKey;
    }



    public void delay(T msg){
        TaskItem<T> task = new TaskItem<T>();
        task.id = UUID.randomUUID().toString();
        task.msg = msg;
        String s = JSON.toJSONString(task);
        jedis.zadd(queueKey, System.currentTimeMillis() + 500, s);
    }

    public void loop(){
        while(!Thread.interrupted()){
            Set<String> values = jedis.zrangeByScore(queueKey, 0, System.currentTimeMillis(), 0,1);

            if(values.isEmpty()){
                try {
                    TimeUnit.MILLISECONDS.sleep(500);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                continue;
            }

           String s =  values.iterator().next();
           if(jedis.zrem(queueKey, s) > 0){

               //反序列化
               TaskItem<T>  task = JSON.parseObject(s, TaskType);

               this.handleMsg(task.msg);
           }
        }
    }


    public void handleMsg(T msg){
        System.out.println(msg);
    }



    public static void main(String [] args){
       Jedis jedis =  RedisUtil.getJedis();

        RedisDelayingQueue<String> queue = new RedisDelayingQueue<>(jedis,"q-demo");



        Thread producer = new Thread(){

          public void run(){
              for(int i = 0 ; i < 10 ; i ++){
                  queue.delay("codehole"+i);
              }
          }

        };

        //lamdba表达式针对于方法级别的
        Thread consumer = new Thread(){

            public void run(){
               queue.loop();
            }

        };

        producer.start();

        consumer.start();

        try{
            producer.join();
            Thread.sleep(6000);
            consumer.join();
        }catch (InterruptedException e){}





    }

}
